package com.chukun.tx.message.task;

import com.chukun.tx.message.queue.DefaultMessageProcessor;
import com.chukun.tx.message.queue.DefaultTimeWheelProcessor;
import com.chukun.tx.message.queue.MessageProcessor;
import com.chukun.tx.message.queue.TimeWheelQueueProcessor;
import com.chukun.tx.model.MqMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author chukun
 * 失败队列的消息写入发送队列
 */
@Component
public class RetryMessageQueueTask{

    private static final Logger LOGGER = LoggerFactory.getLogger(RetryMessageQueueTask.class);

    private ExecutorService executorService = new ThreadPoolExecutor(1,2,10, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(10), r -> {
        Thread t = new Thread(r);
        t.setName("retryMessageQueueTask thread");
        t.setDaemon(true);
        return t;
    }, new ThreadPoolExecutor.CallerRunsPolicy());

    @PostConstruct
    public void handleFailMsg() {
        RetryMessageRunnable retryRun = new RetryMessageRunnable();
        for ( ; ;) {
            executorService.execute(retryRun);
        }
    }

    /**
     * 放入到待发送队列
     */
    private static class RetryMessageRunnable implements Runnable {

        private TimeWheelQueueProcessor timeWheelQueueProcessor = DefaultTimeWheelProcessor.getDefaultProcessor();

        private MessageProcessor messageProcessor = DefaultMessageProcessor.getDefaultProcessor();

        @Override
        public void run() {
            try {
                long curTime = System.currentTimeMillis();
                MqMessage message = timeWheelQueueProcessor.lookTopMsg();
                while (message != null && message.getNextExpireTime() <= curTime) {
                    message = timeWheelQueueProcessor.getFailMessage();
                    messageProcessor.putMsg(message);
                }
            }catch (Exception e) {
                LOGGER.error("pool message error ", e);
            }
        }
    }
}
